package consumer

import (
	"github.com/Shopify/sarama"
	"log"
	"strings"
)

type KafkaConsumer interface {
	Init()
	Close() error
}

func New(conf Config, receiveChan chan<- []byte) KafkaConsumer {
	return &consumer{
		Config:      conf,
		receiveChan: receiveChan,
		closeSignal: make(chan bool),
	}
}

type consumer struct {
	client      sarama.Client
	Consumer    sarama.Consumer
	receiveChan chan<- []byte
	closeSignal chan bool
	Config
}
type Config struct {
	Hosts      string
	Topic      string
	Partition  int
	SaslEnable bool
	UserName   string
	Password   string
}

func (this *consumer) Init() {
	log.Println("启动kafka消费者")
	if err := this.newConsumer(); err != nil {
		panic(err.Error())
	}
	go func() {
		this.waitReadMsg()
	}()
	log.Println("kafka消费者已启动")
}

//会阻塞，使用协程
func (this *consumer) waitReadMsg() {

	partitionConsumer, err := this.Consumer.ConsumePartition(this.Topic, int32(this.Partition), sarama.OffsetNewest)
	if err != nil {
		panic(err.Error())
	}

	defer this.Consumer.Close()
	go func() {
		log.Println("kafka生产者消息等待中")
		defer log.Println("kafka生产者消息等待退出")
		for {
			select {
			case msg := <-partitionConsumer.Messages():
				//log.Printf("Consumed message: [%s], offset: [%d]\n", msg.Value, msg.Offset)
				this.receiveChan <- msg.Value
			case <-this.closeSignal:
				return
			}
		}
	}()
}

func (this *consumer) Close() error {
	defer log.Println("kafka消费者退出")
	this.closeSignal <- true
	return nil
}

func (this *consumer) newConsumer() error {
	var err error
	//创建配置
	config := sarama.NewConfig()
	if this.SaslEnable {
		config.Net.SASL.Enable = true
		config.Net.SASL.User = this.UserName
		config.Net.SASL.Password = this.Password
	}
	//创建客户端
	client, err := sarama.NewClient(strings.Split(this.Hosts, ","), config)
	if err != nil {
		return err
	}
	//创建消费者
	this.Consumer, err = sarama.NewConsumerFromClient(client)
	return err
}
